package com.google.code.p.restrpc.client.proxy;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.UUID;

import org.apache.commons.lang.StringUtils;

import com.google.code.p.restrpc.AsyncRpcResult;
import com.google.code.p.restrpc.MutableRpcRequest;
import com.google.code.p.restrpc.RestInvoker;
import com.google.code.p.restrpc.RestInvokerFactory;
import com.google.code.p.restrpc.RpcInvoker;
import com.google.code.p.restrpc.RpcOptions;
import com.google.code.p.restrpc.RpcProtocal;
import com.google.code.p.restrpc.RpcRequest;
import com.google.code.p.restrpc.client.AsyncRpcCallback;
import com.google.code.p.restrpc.client.AsyncRpcResultQuery;
import com.google.code.p.restrpc.client.DefaultRpcContext;
import com.google.code.p.restrpc.client.DefaultRpcOptions;
import com.google.code.p.restrpc.client.RpcOptionsFactory;
import com.google.code.p.restrpc.client.RpcProxy;
import com.google.code.p.restrpc.client.ServiceNotRegisteredException;
import com.google.code.p.restrpc.client.cache.ServiceMetadataRepository;
import com.google.code.p.restrpc.client.cache.ServiceMetadataRepositoryProvider;
import com.google.code.p.restrpc.client.event.AsyncRpcEventListener;
import com.google.code.p.restrpc.client.event.AsyncRpcEventObject;
import com.google.code.p.restrpc.client.registry.CachedRegistryClientFactory;
import com.google.code.p.restrpc.client.registry.DefaultRegistryClientFactory;
import com.google.code.p.restrpc.client.registry.RegistryClient;
import com.google.code.p.restrpc.client.registry.RegistryClientFactory;
import com.google.code.p.restrpc.core.url.AbstractRpcRequest;
import com.google.code.p.restrpc.core.url.DefaultRestfulRequest;
import com.google.code.p.restrpc.core.url.DefaultSecureRestfulRequest;
import com.google.code.p.restrpc.directory.RemoteServiceDefinition;
import com.google.code.p.restrpc.directory.RemoteServiceDefinitionPack;
import com.google.code.p.restrpc.event.RpcEventCenter;
import com.google.code.p.restrpc.event.RpcEventObject;

/**
 * @author dz
 */
public final class RpcProxyFactory implements
                                  RpcProxy,
                                  RestInvokerFactory,
                                  RegistryClientFactory,
                                  RpcProxyMetadata {
    
    // ***************************************************
    // static helper
    // ***************************************************
    private static ThreadLocal<String> rpcInvokeIdHolder = new ThreadLocal<String>();
    
    private static ThreadLocal<AsyncRpcResultQuery> asyncRpcResultQueryHolder = new ThreadLocal<AsyncRpcResultQuery>();
    
    static {
        RpcEventCenter.getInstance()
                      .registerListener(new AsyncRpcEventListener() {
                          
                          public boolean support(RpcEventObject eventObject) {
                              return (eventObject instanceof AsyncRpcEventObject);
                          }
                          
                          public void handle(RpcEventObject eventObject) {
                              final AsyncRpcEventObject asyncEventObject = (AsyncRpcEventObject) eventObject;
                              asyncRpcResultQueryHolder.set(new AsyncRpcResultQuery() {
                                  public String getRequestedId() {
                                      return asyncEventObject.getAsyncRequestId();
                                  }
                                  
                                  public String getQueryUrl() {
                                      return asyncEventObject.getQueryUrl();
                                  }
                                  
                                  public AsyncRpcResult queryResult() {
                                      if (StringUtils.isEmpty(asyncEventObject.getAsyncRequestId())) {
                                          return null;
                                      }
                                      
                                      if (asyncEventObject.getAsyncRpcResultQuery() == null) {
                                          return null;
                                      }
                                      
                                      return asyncEventObject.getAsyncRpcResultQuery()
                                                             .findById(asyncEventObject.getAsyncRequestId());
                                  }
                              });
                          }
                          
                      });
    }
    
    public static AsyncRpcResultQuery invoke(AsyncRpcCallback callback) {
        final String id = UUID.randomUUID().toString();
        try {
            rpcInvokeIdHolder.set(id);
            callback.execute();
            return asyncRpcResultQueryHolder.get();
        }
        finally {
            rpcInvokeIdHolder.set(null);
            asyncRpcResultQueryHolder.set(null);
        }
    }
    
    static String getCurrentAsyncRpcId() {
        return rpcInvokeIdHolder.get();
    }
    
    // ***************************************************
    // static create helper
    // ***************************************************
    private static class RpcProxyFactoryHolder {
        static final RpcProxyFactory instance;
        
        static {
            instance = new RpcProxyFactory(RpcOptionsFactory.getInstance());
            if (!StringUtils.isEmpty(RpcOptionsFactory.getInstance()
                                                      .getDirectoryServiceUrl())) {
                instance.setRegistryClientFactory(new CachedRegistryClientFactory(new DefaultRegistryClientFactory()));
            }
        }
    }
    
    /**
     * @return always return the same singlton instance of RpcProxy
     */
    public static RpcProxy getInstance() {
        return RpcProxyFactoryHolder.instance;
    }
    
    /**
     * @param properties
     * @return retuen new instance
     */
    public static RpcProxy newInstance(Properties properties) {
        return newInstance(new DefaultRpcOptions(properties));
    }
    
    /**
     * @param rpcOptions
     * @return retuen new instance
     */
    public static RpcProxy newInstance(RpcOptions rpcOptions) {
        RpcProxyFactory result = new RpcProxyFactory(rpcOptions);
        if (!StringUtils.isEmpty(rpcOptions.getDirectoryServiceUrl())) {
            result.setRegistryClientFactory(new CachedRegistryClientFactory(new DefaultRegistryClientFactory(rpcOptions.getDirectoryServiceUrl()),
                                                                            rpcOptions.getHeartbeat()));
        }
        return result;
    }
    
    // ***************************************************
    // instance field definition
    // ***************************************************
    Date createDateTime = new Date();
    
    RpcOptions rpcOptions;
    
    boolean useJdkProxy = true;
    
    RestInvokerFactory restInvokerFactory = new CachedRpcInvokerFactory(new RestfulRpcInvokerFactory());
    
    RegistryClientFactory registryClientFactory;
    
    // ***************************************************
    // constructor definition
    // ***************************************************
    /**
     * @param options
     */
    RpcProxyFactory(RpcOptions options) {
        this.rpcOptions = options;
        RpcProxyMetadataTracker.getInstance().registerRpcProxyMetadata(this);
    }
    
    // ***************************************************
    // instance method definition
    // ***************************************************
    public Date getCreateDateTime() {
        return this.createDateTime;
    }
    
    /**
     * @return the rpcOptions
     */
    public RpcOptions getRpcOptions() {
        return rpcOptions;
    }
    
    /**
     * 
     */
    public ServiceMetadataRepository getServiceMetadataRepository() {
        if (registryClientFactory != null) {
            if (registryClientFactory instanceof ServiceMetadataRepositoryProvider) {
                return ((ServiceMetadataRepositoryProvider) registryClientFactory).getServiceMetadataRepository();
            }
        }
        return null;
    }
    
    /**
     * @return the useJdkProxy
     */
    public boolean isUseJdkProxy() {
        return useJdkProxy;
    }
    
    /**
     * @param useJdkProxy
     *            the useJdkProxy to set
     */
    public void setUseJdkProxy(boolean useJdkProxy) {
        this.useJdkProxy = useJdkProxy;
    }
    
    /**
     * @return
     */
    public RestInvoker getRestInvoker() {
        return restInvokerFactory.getRestInvoker();
    }
    
    /**
     * @param restInvokerFactory
     */
    public void setRestInvokerFactory(RestInvokerFactory restInvokerFactory) {
        if (restInvokerFactory == null) {
            throw new IllegalArgumentException("The 'restInvokerFactory' must not be null!");
        }
        this.restInvokerFactory = restInvokerFactory;
    }
    
    /**
     * @return the restInvokerFactory
     */
    public RestInvokerFactory getRestInvokerFactory() {
        return restInvokerFactory;
    }
    
    /**
     * @return
     */
    public RegistryClientFactory getRegistryClientFactory() {
        return registryClientFactory;
    }
    
    /**
     * @return
     */
    public RegistryClient getRegistryClient() {
        return registryClientFactory.getRegistryClient();
    }
    
    /**
     * 
     */
    public RegistryClient getRegistryClient(String directoryServiceUrl) {
        return registryClientFactory.getRegistryClient(directoryServiceUrl);
    }
    
    /**
     * @param registryClientFactory
     */
    public void setRegistryClientFactory(RegistryClientFactory registryClientFactory) {
        if (registryClientFactory == null) {
            throw new IllegalArgumentException("The 'registryClientFactory' must not be null!");
        }
        this.registryClientFactory = registryClientFactory;
    }
    
    /**
     * 
     */
    public <T> T createProxy(final Class<T> serviceInterface, final String url) {
        return createProxy(serviceInterface.getClassLoader(),
                           serviceInterface,
                           url,
                           rpcOptions);
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param serviceName
     * @return
     */
    public <T> T createProxy(final ClassLoader classLoader,
                             final Class<T> serviceInterface,
                             final String url) {
        return createProxy(serviceInterface.getClassLoader(),
                           serviceInterface,
                           url,
                           rpcOptions);
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param serviceName
     * @return
     */
    private <T> T createProxy(final ClassLoader classLoader,
                              final Class<T> serviceInterface,
                              final String url,
                              final RpcOptions rpcOptions) {
        if (serviceInterface == null) {
            throw new IllegalArgumentException("The 'serviceInterface' can't be null!");
        }
        if (StringUtils.isEmpty(url)) {
            throw new IllegalArgumentException("The 'url' can't be empty!");
        }
        
        List<RpcRequest> rpcRequests = new ArrayList<RpcRequest>();
        
        if (RpcProtocal.HTTP.isSupported(url) || RpcProtocal.HTTPS.isSupported(url)) {
            RpcRequest[] requests = rpcOptions.getUrlParser().parse(url);
            for (RpcRequest rpcRequest : requests) {
                if (rpcRequest instanceof MutableRpcRequest) {
                    ((MutableRpcRequest) rpcRequest).setTransferFormat(rpcOptions.getTransferFormat());
                }
                
                rpcRequests.add(rpcRequest);
            }
        }
        else if (RpcProtocal.LOOKUP.isSupported(url) || RpcProtocal.LOOKUPS.isSupported(url)) {
            if (getRegistryClientFactory() == null) {
                throw new IllegalArgumentException("Protocal 'lookup' or 'lookups' is not supported!Please make sure the 'registryClientFactory' is provided!");
            }
            
            String serviceName = null;
            if (RpcProtocal.LOOKUPS.isSupported(url)) {
                serviceName = url.substring(RpcProtocal.LOOKUPS.prefix()
                                                               .length());
            }
            else {
                serviceName = url.substring(RpcProtocal.LOOKUP.prefix()
                                                              .length());
            }
            if (StringUtils.isEmpty(serviceName)) {
                throw new IllegalArgumentException("Illegal url " + url);
            }
            
            RemoteServiceDefinitionPack serviceDefinitionsPack = registryClientFactory.getRegistryClient()
                                                                                      .lookup(serviceName,
                                                                                              -1l);
            
            if (serviceDefinitionsPack == null) {
                throw new ServiceNotRegisteredException(serviceName);
            }
            
            for (RemoteServiceDefinition serviceDefinition : serviceDefinitionsPack.getRemoteServiceDefinitions()) {
                AbstractRpcRequest request = null;
                if (serviceDefinition.isTlsEnabled()) {
                    request = new DefaultSecureRestfulRequest();
                }
                else {
                    request = new DefaultRestfulRequest();
                }
                
                request.setTransferFormat(rpcOptions.getTransferFormat());
                request.setUrl(serviceDefinition.toServiceUrl());
                request.setAddress(serviceDefinition.getRemoteAddress());
                request.setPort(serviceDefinition.getRemotePort());
                request.setContext(serviceDefinition.getContextPath());
                request.setWeight(serviceDefinition.getWeight());
                request.setServiceName(serviceName);
                rpcRequests.add(request);
            }
            
            if (rpcRequests.size() == 0) {
                throw new ServiceNotRegisteredException(serviceName);
            }
            
            return createCachedRpcProxy(classLoader,
                                        serviceInterface,
                                        rpcRequests,
                                        rpcOptions);
        }
        else if (url.indexOf(':') > 0) {
            throw new IllegalArgumentException("Unsupported Rpc Protocal " + url.substring(0,
                                                                                           url.indexOf(':')));
        }
        else {
            throw new IllegalArgumentException("Unsupported url" + url);
        }
        
        return createRpcProxy(classLoader,
                              serviceInterface,
                              rpcRequests,
                              rpcOptions);
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param request
     * @param rpcOptions
     * @return
     */
    private <T> T createRpcProxy(final ClassLoader classLoader,
                                 final Class<T> serviceInterface,
                                 final List<RpcRequest> requests,
                                 final RpcOptions rpcOptions) {
        if ((requests == null) || requests.isEmpty()) {
            throw new IllegalArgumentException("No requests specified!");
        }
        if (requests.size() == 1) {
            return (T) newDefaultRpcProxy(classLoader,
                                          serviceInterface,
                                          requests.get(0));
        }
        
        return (T) newClusteredRpcProxy(classLoader,
                                        serviceInterface,
                                        requests,
                                        rpcOptions);
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param request
     * @param rpcOptions
     * @return
     */
    private <T> T createCachedRpcProxy(final ClassLoader classLoader,
                                       final Class<T> serviceInterface,
                                       final List<RpcRequest> requests,
                                       final RpcOptions rpcOptions) {
        if ((requests == null) || requests.isEmpty()) {
            throw new IllegalArgumentException("No requests specified!");
        }
        if (requests.size() == 1) {
            
            return (T) newDefaultRpcProxy(classLoader,
                                          serviceInterface,
                                          requests.get(0));
        }
        
        return (T) newCachedClusteredRpcProxy(classLoader,
                                              serviceInterface,
                                              requests,
                                              rpcOptions);
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param rpcRequest
     * @return
     */
    private <T> T newDefaultRpcProxy(final ClassLoader classLoader,
                                     final Class<T> serviceInterface,
                                     final RpcRequest rpcRequest) {
        final DefaultRpcContext rpcContext = new DefaultRpcContext(this);
        rpcContext.setServiceName(rpcRequest.getServiceName());
        rpcContext.setTargetClass(serviceInterface);
        return (T) Proxy.newProxyInstance(classLoader,
                                          new Class[] { serviceInterface },
                                          new InvocationHandler() {
                                              
                                              public Object invoke(Object proxy,
                                                                   Method method,
                                                                   Object[] args) throws Throwable {
                                                  RpcInvoker rpcInvoker = null;
                                                  
                                                  if (RpcProxyFactory.getCurrentAsyncRpcId() != null) {
                                                      rpcInvoker = new DefaultAsyncRpcInvoker(getRestInvoker());
                                                  }
                                                  else {
                                                      rpcInvoker = new DefaultRpcInvoker(getRestInvoker());
                                                  }
                                                  
                                                  if ("toString".equals(method.getName())) {
                                                      return rpcInvoker.toString();
                                                  }
                                                  
                                                  if ("hashCode".equals(method.getName())) {
                                                      return rpcInvoker.hashCode();
                                                  }
                                                  
                                                  if ("equals".equals(method.getName())) {
                                                      return rpcInvoker.equals(args[0]);
                                                  }
                                                  
                                                  rpcContext.setTargetMethod(method);
                                                  rpcContext.setParameters(args);
                                                  
                                                  rpcRequest.setBody(args);
                                                  ((MutableRpcRequest) rpcRequest).setMethodName(method.getName());
                                                  
                                                  if (RpcProxyFactory.getCurrentAsyncRpcId() != null) {
                                                      ((MutableRpcRequest) rpcRequest).setAsync(true);
                                                      ((MutableRpcRequest) rpcRequest).setRequestId(RpcProxyFactory.getCurrentAsyncRpcId());
                                                  }
                                                  
                                                  return rpcInvoker.invoke(rpcContext,
                                                                           rpcRequest);
                                              }
                                              
                                          });
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param request
     * @param rpcOptions
     * @return
     */
    private <T> T newClusteredRpcProxy(final ClassLoader classLoader,
                                       final Class<T> serviceInterface,
                                       final List<RpcRequest> requests,
                                       final RpcOptions rpcOptions) {
        final DefaultRpcContext rpcContext = new DefaultRpcContext(this,
                                                                   rpcOptions);
        rpcContext.setTargetClass(serviceInterface);
        rpcContext.setServiceName(requests.get(0).getServiceName());
        return (T) Proxy.newProxyInstance(classLoader,
                                          new Class[] { serviceInterface },
                                          new InvocationHandler() {
                                              
                                              public Object invoke(Object proxy,
                                                                   Method method,
                                                                   Object[] args) throws Throwable {
                                                  RpcInvoker rpcInvoker = null;
                                                  
                                                  if (RpcProxyFactory.getCurrentAsyncRpcId() != null) {
                                                      rpcInvoker = new ClusteredAsyncRpcInvoker(getRestInvoker());
                                                  }
                                                  else {
                                                      rpcInvoker = new ClusteredRpcInvoker(getRestInvoker());
                                                  }
                                                  
                                                  if ("toString".equals(method.getName())) {
                                                      return rpcInvoker.toString();
                                                  }
                                                  
                                                  if ("hashCode".equals(method.getName())) {
                                                      return rpcInvoker.hashCode();
                                                  }
                                                  
                                                  if ("equals".equals(method.getName())) {
                                                      return rpcInvoker.equals(args[0]);
                                                  }
                                                  
                                                  rpcContext.setTargetMethod(method);
                                                  rpcContext.setParameters(args);
                                                  
                                                  for (RpcRequest rpcRequest : requests) {
                                                      rpcRequest.setBody(args);
                                                      ((MutableRpcRequest) rpcRequest).setMethodName(method.getName());
                                                      
                                                      if (RpcProxyFactory.getCurrentAsyncRpcId() != null) {
                                                          ((MutableRpcRequest) rpcRequest).setAsync(true);
                                                          ((MutableRpcRequest) rpcRequest).setRequestId(RpcProxyFactory.getCurrentAsyncRpcId());
                                                      }
                                                  }
                                                  
                                                  return rpcInvoker.invoke(rpcContext,
                                                                           requests);
                                              }
                                              
                                          });
    }
    
    /**
     * @param <T>
     * @param classLoader
     * @param serviceInterface
     * @param request
     * @param rpcOptions
     * @return
     */
    private <T> T newCachedClusteredRpcProxy(final ClassLoader classLoader,
                                             final Class<T> serviceInterface,
                                             final List<RpcRequest> requests,
                                             final RpcOptions rpcOptions) {
        final DefaultRpcContext rpcContext = new DefaultRpcContext(this,
                                                                   rpcOptions);
        rpcContext.setTargetClass(serviceInterface);
        rpcContext.setServiceName(requests.get(0).getServiceName());
        return (T) Proxy.newProxyInstance(classLoader,
                                          new Class[] { serviceInterface },
                                          new InvocationHandler() {
                                              
                                              public Object invoke(Object proxy,
                                                                   Method method,
                                                                   Object[] args) throws Throwable {
                                                  RpcInvoker rpcInvoker = null;
                                                  
                                                  if (RpcProxyFactory.getCurrentAsyncRpcId() != null) {
                                                      rpcInvoker = new CachedClusteredAsyncRpcInvoker(getRestInvoker());
                                                  }
                                                  else {
                                                      rpcInvoker = new CachedClusteredRpcInvoker(getRestInvoker());
                                                  }
                                                  
                                                  if ("toString".equals(method.getName())) {
                                                      return ((Object) rpcInvoker).toString();
                                                  }
                                                  
                                                  if ("hashCode".equals(method.getName())) {
                                                      return ((Object) rpcInvoker).hashCode();
                                                  }
                                                  
                                                  if ("equals".equals(method.getName())) {
                                                      return ((Object) rpcInvoker).equals(args[0]);
                                                  }
                                                  
                                                  rpcContext.setTargetMethod(method);
                                                  rpcContext.setParameters(args);
                                                  
                                                  for (RpcRequest rpcRequest : requests) {
                                                      rpcRequest.setBody(args);
                                                      ((MutableRpcRequest) rpcRequest).setMethodName(method.getName());
                                                      
                                                      if (RpcProxyFactory.getCurrentAsyncRpcId() != null) {
                                                          ((MutableRpcRequest) rpcRequest).setAsync(true);
                                                          ((MutableRpcRequest) rpcRequest).setRequestId(RpcProxyFactory.getCurrentAsyncRpcId());
                                                      }
                                                  }
                                                  
                                                  return rpcInvoker.invoke(rpcContext,
                                                                           requests);
                                              }
                                              
                                          });
    }
    
}
